package org.apache.rocketmq.samples.springboot;

import com.alibaba.fastjson.TypeReference;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.samples.springboot.domain.OrderPaidEvent;
import org.apache.rocketmq.samples.springboot.domain.ProductWithPayload;
import org.apache.rocketmq.samples.springboot.domain.User;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalRequestCallback;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessageHeaders;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.util.MimeTypeUtils;

import javax.annotation.Resource;
import java.math.BigDecimal;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 生产者，使用RocketMQTemplate发送各种类型的消息
 */
@SpringBootApplication
public class ProducerApplication implements CommandLineRunner {
    
    // RocketMQ配置属性
    @Resource
    private RocketMQTemplate rocketMQTemplate;
    
    @Value("${demo.rocketmq.transTopic}")
    private String springTransTopic;
    
    @Value("${demo.rocketmq.topic}")
    private String springTopic;
    
    @Value("${demo.rocketmq.topic.user}")
    private String userTopic;

    @Value("${demo.rocketmq.orderTopic}")
    private String orderPaidTopic;
    
    @Value("${demo.rocketmq.msgExtTopic}")
    private String msgExtTopic;
    
    @Value("${demo.rocketmq.stringRequestTopic}")
    private String stringRequestTopic;
    
    @Value("${demo.rocketmq.bytesRequestTopic}")
    private String bytesRequestTopic;
    
    @Value("${demo.rocketmq.objectRequestTopic}")
    private String objectRequestTopic;
    
    @Value("${demo.rocketmq.genericRequestTopic}")
    private String genericRequestTopic;

    @Resource(name = "extRocketMQTemplate")
    private RocketMQTemplate extRocketMQTemplate;

    public static void main(String[] args) {
        SpringApplication.run(ProducerApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        try {
            // 1. 发送普通消息
            sendNormalMessages();
            
            // 2. 发送批量消息
            sendBatchMessages();
            
            // 3. 发送事务消息
            sendTransactionalMessages();
            
            // 4. 发送请求-回复模式消息
            sendRequestReplyMessages();
        } catch (Exception e) {
            System.err.println("执行消息发送时发生错误: " + e.getMessage());
            e.printStackTrace();
        }
    }
    
    /**
     * 发送普通消息
     */
    private void sendNormalMessages() throws Exception {
        // 发送字符串消息
        SendResult sendResult = rocketMQTemplate.syncSend(springTopic, "Hello, World!");
        System.out.printf("同步发送1到主题 %s 发送结果=%s %n", springTopic, sendResult);

        // 发送对象消息
        sendResult = rocketMQTemplate.syncSend(userTopic, new User().setUserAge((byte) 18).setUserName("Kitty"));
        System.out.printf("同步发送1到主题 %s 发送结果=%s %n", userTopic, sendResult);

        // 发送带JSON头信息的用户对象消息
        sendResult = rocketMQTemplate.syncSend(userTopic, MessageBuilder.withPayload(
                new User().setUserAge((byte) 21).setUserName("Lester")).setHeader(MessageHeaders.CONTENT_TYPE, MimeTypeUtils.APPLICATION_JSON_VALUE).build());
        System.out.printf("同步发送1到主题 %s 发送结果=%s %n", userTopic, sendResult);

        // 使用Spring Message发送字节数组消息
        sendResult = extRocketMQTemplate.syncSend(springTopic, MessageBuilder.withPayload("Hello, World!2222".getBytes()).build());
        System.out.printf("扩展RocketMQTemplate同步发送1到主题 %s 发送结果=%s %n", springTopic, sendResult);

        // 使用Spring Message发送字符串消息
        sendResult = rocketMQTemplate.syncSend(springTopic, MessageBuilder.withPayload("Hello, World! I'm from spring message").build());
        System.out.printf("同步发送2到主题 %s 发送结果=%s %n", springTopic, sendResult);

        // 异步发送订单支付事件消息，通过回调处理发送结果
        rocketMQTemplate.asyncSend(orderPaidTopic, new OrderPaidEvent("T_001", new BigDecimal("88.00")), new SendCallback() {
            @Override
            public void onSuccess(SendResult var1) {
                System.out.printf("异步发送成功 发送结果=%s %n", var1);
            }

            @Override
            public void onException(Throwable var1) {
                System.out.printf("异步发送异常 异常信息=%s %n", var1);
            }

        });

        // 发送带特殊标签的消息
        rocketMQTemplate.convertAndSend(msgExtTopic + ":tag0", "I'm from tag0");
        System.out.printf("同步发送主题 %s 标签 %s %n", msgExtTopic, "tag0");
        // 发送带特殊标签的消息
        rocketMQTemplate.convertAndSend(msgExtTopic + ":tag1", "I'm from tag1");
        System.out.printf("同步发送主题 %s 标签 %s %n", msgExtTopic, "tag1");
    }
    
    /**
     * 发送批量消息
     */
    private void sendBatchMessages() {
        // 发送批量字符串消息
        testBatchMessages();

        // 有序发送批量字符串消息
        testSendBatchMessageOrderly();
    }
    
    /**
     * 发送事务消息
     */
    private void sendTransactionalMessages() throws MessagingException {
        // 使用rocketMQTemplate发送事务消息
        testRocketMQTemplateTransaction();

        // 使用extRocketMQTemplate发送事务消息
        testExtRocketMQTemplateTransaction();
    }
    
    /**
     * 发送请求-回复模式消息
     */
    private void sendRequestReplyMessages() {
        // === 同步请求-回复模式 ===
        
        // 同步发送字符串请求并等待回复
        try {
            // 以同步模式发送请求并接收字符串类型的回复
            String replyString = rocketMQTemplate.sendAndReceive(stringRequestTopic, "request string", String.class, 5000);
            System.out.printf("发送 %s 并接收 %s %n", "request string", replyString);
        } catch (Exception e) {
            System.err.printf("发送字符串请求消息失败: %s%n", e.getMessage());
            System.out.println("请确保StringConsumerWithReplyString消费者服务正在运行");
        }

        // 同步发送字节请求并等待回复
        try {
            // 以同步模式发送请求（带超时参数）并接收字节数组类型的回复
            byte[] replyBytes = rocketMQTemplate.sendAndReceive(bytesRequestTopic, MessageBuilder.withPayload("request byte[]").build(), byte[].class, 3000);
            System.out.printf("发送 %s 并接收 %s %n", "request byte[]", new String(replyBytes));
        } catch (Exception e) {
            System.err.printf("发送字节请求消息失败: %s%n", e.getMessage());
            System.out.println("请确保ConsumerWithReplyBytes消费者服务正在运行");
        }

        // 同步发送用户对象请求并等待回复
        try {
            // 以同步模式发送请求（带hashKey参数）并接收User类型的回复
            User requestUser = new User().setUserAge((byte) 9).setUserName("requestUserName");
            User replyUser = rocketMQTemplate.sendAndReceive(objectRequestTopic, requestUser, User.class, "order-id", 5000);
            System.out.printf("发送 %s 并接收 %s %n", requestUser, replyUser);
        } catch (Exception e) {
            System.err.printf("发送用户对象请求消息失败: %s%n", e.getMessage());
            System.out.println("请确保ObjectConsumerWithReplyUser消费者服务正在运行");
        }
        
        // 同步发送泛型请求并等待回复
        try {
            // 以同步模式发送请求（带超时和延迟级别参数）并接收泛型类型的回复
            ProductWithPayload<String> replyGenericObject = rocketMQTemplate.sendAndReceive(genericRequestTopic, "request generic",
                    new TypeReference<ProductWithPayload<String>>() {
                    }.getType(), 50000, 2);
            System.out.printf("发送 %s 并接收 %s %n", "request generic", replyGenericObject);
        } catch (Exception e) {
            System.err.printf("发送泛型请求消息失败: %s%n", e.getMessage());
            System.out.println("请确保ConsumerWithReplyGeneric消费者服务正在运行");
        }

        // === 异步请求-回复模式 ===
        
        // 异步发送字符串请求并等待回复
        try {
            // 以异步模式发送请求并接收字符串类型的回复
            rocketMQTemplate.sendAndReceive(stringRequestTopic, "request string", new RocketMQLocalRequestCallback<String>() {
                @Override
                public void onSuccess(String message) {
                    System.out.printf("发送 %s 并接收 %s %n", "request string", message);
                }

                @Override
                public void onException(Throwable e) {
                    System.err.printf("异步发送字符串请求消息失败: %s%n", e.getMessage());
                    System.out.println("请确保StringConsumerWithReplyString消费者服务正在运行");
                }
            }, 5000);
        } catch (Exception e) {
            System.err.printf("异步发送字符串请求消息失败: %s%n", e.getMessage());
            System.out.println("请确保StringConsumerWithReplyString消费者服务正在运行");
        }

        // 异步发送用户对象请求并等待回复
        try {
            // 以异步模式发送请求并接收User类型的回复
            rocketMQTemplate.sendAndReceive(objectRequestTopic, new User().setUserAge((byte) 9).setUserName("requestUserName"), new RocketMQLocalRequestCallback<User>() {
                @Override
                public void onSuccess(User message) {
                    System.out.printf("发送用户对象并接收 %s %n", message.toString());
                }

                @Override
                public void onException(Throwable e) {
                    System.err.printf("异步发送用户对象请求消息失败: %s%n", e.getMessage());
                    System.out.println("请确保ObjectConsumerWithReplyUser消费者服务正在运行");
                }
            }, 5000);
        } catch (Exception e) {
            System.err.printf("异步发送用户对象请求消息失败: %s%n", e.getMessage());
            System.out.println("请确保ObjectConsumerWithReplyUser消费者服务正在运行");
        }
    }

    /**
     * 测试批量消息发送
     */
    private void testBatchMessages() {
        List<Message> msgs = new ArrayList<Message>();
        for (int i = 0; i < 10; i++) {
            msgs.add(MessageBuilder.withPayload("Hello RocketMQ Batch Msg#" + i).
                    setHeader(RocketMQHeaders.KEYS, "KEY_" + i).build());
        }

        SendResult sr = rocketMQTemplate.syncSend(springTopic, msgs, 60000);

        System.out.printf("--- 批量消息发送结果 :" + sr);
    }

    /**
     * 测试有序批量消息发送
     */
    private void testSendBatchMessageOrderly() {
        for (int q = 0; q < 4; q++) {
            // 发送到4个队列
            List<Message> msgs = new ArrayList<Message>();
            for (int i = 0; i < 10; i++) {
                int msgIndex = q * 10 + i;
                String msg = String.format("Hello RocketMQ Batch Msg#%d to queue: %d", msgIndex, q);
                msgs.add(MessageBuilder.withPayload(msg).
                        setHeader(RocketMQHeaders.KEYS, "KEY_" + msgIndex).build());
            }
            SendResult sr = rocketMQTemplate.syncSendOrderly(springTopic, msgs, q + "", 60000);
            System.out.println("--- 批量消息有序发送到队列 :" + sr.getMessageQueue().getQueueId() + " 发送结果 :" + sr);
        }
    }

    /**
     * 测试使用rocketMQTemplate发送事务消息
     */
    private void testRocketMQTemplateTransaction() throws MessagingException {
        String[] tags = new String[]{"TagA", "TagB", "TagC", "TagD", "TagE"};
        for (int i = 0; i < 10; i++) {
            try {

                Message msg = MessageBuilder.withPayload("rocketMQTemplate事务消息 " + i).
                        setHeader(RocketMQHeaders.TRANSACTION_ID, "KEY_" + i).build();
                SendResult sendResult = rocketMQTemplate.sendMessageInTransaction(
                        springTransTopic + ":" + tags[i % tags.length], msg, null);
                System.out.printf("------rocketMQTemplate发送事务消息体 = %s , 发送结果=%s %n",
                        msg.getPayload(), sendResult.getSendStatus());

                Thread.sleep(10);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 测试使用extRocketMQTemplate发送事务消息
     */
    private void testExtRocketMQTemplateTransaction() throws MessagingException {
        for (int i = 0; i < 10; i++) {
            try {
                Message msg = MessageBuilder.withPayload("extRocketMQTemplate事务消息 " + i).
                        setHeader(RocketMQHeaders.TRANSACTION_ID, "KEY_" + i).build();
                SendResult sendResult = extRocketMQTemplate.sendMessageInTransaction(
                        springTransTopic, msg, null);
                System.out.printf("------扩展RocketMQTemplate发送事务消息体 = %s , 发送结果=%s %n",
                        msg.getPayload(), sendResult.getSendStatus());

                Thread.sleep(10);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 事务监听器实现
     */
    @RocketMQTransactionListener
    class TransactionListenerImpl implements RocketMQLocalTransactionListener {
        private AtomicInteger transactionIndex = new AtomicInteger(0);

        private ConcurrentHashMap<String, Integer> localTrans = new ConcurrentHashMap<String, Integer>();

        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
            System.out.printf("#### executeLocalTransaction被执行, 消息事务ID=%s %n",
                    transId);
            int value = transactionIndex.getAndIncrement();
            int status = value % 3;
            localTrans.put(transId, status);
            if (status == 0) {
                // 返回本地事务成功（提交），在这种情况下，
                // 此消息将不会在checkLocalTransaction()中被检查
                System.out.printf("    # COMMIT # 模拟消息 %s 相关本地事务执行成功! ### %n", msg.getPayload());
                return RocketMQLocalTransactionState.COMMIT;
            }

            if (status == 1) {
                // 返回本地事务失败（回滚），在这种情况下，
                // 此消息将不会在checkLocalTransaction()中被检查
                System.out.printf("    # ROLLBACK # 模拟 %s 相关本地事务执行失败! %n", msg.getPayload());
                return RocketMQLocalTransactionState.ROLLBACK;
            }

            System.out.printf("    # UNKNOW # 模拟 %s 相关本地事务执行状态未知! \n", msg.getPayload());
            return RocketMQLocalTransactionState.UNKNOWN;
        }

        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            String transId = (String) msg.getHeaders().get(RocketMQHeaders.TRANSACTION_ID);
            RocketMQLocalTransactionState retState = RocketMQLocalTransactionState.COMMIT;
            Integer status = localTrans.get(transId);
            if (null != status) {
                switch (status) {
                    case 0:
                        retState = RocketMQLocalTransactionState.COMMIT;
                        break;
                    case 1:
                        retState = RocketMQLocalTransactionState.ROLLBACK;
                        break;
                    case 2:
                        retState = RocketMQLocalTransactionState.UNKNOWN;
                        break;
                }
            }
            System.out.printf("------ !!! checkLocalTransaction被执行一次," +
                            " 消息事务ID=%s, 事务状态=%s 状态=%s %n",
                    transId, retState, status);
            return retState;
        }
    }

    /**
     * 扩展事务监听器实现
     */
    @RocketMQTransactionListener(rocketMQTemplateBeanName = "extRocketMQTemplate")
    class ExtTransactionListenerImpl implements RocketMQLocalTransactionListener {
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            System.out.printf("ExtTransactionListenerImpl executeLocalTransaction并返回UNKNOWN。 \n");
            return RocketMQLocalTransactionState.UNKNOWN;
        }

        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            System.out.printf("ExtTransactionListenerImpl checkLocalTransaction并返回COMMIT。 \n");
            return RocketMQLocalTransactionState.COMMIT;
        }
    }
}